package com.tech.consumer

import com.tech.common.{KuduUtil, MysqlUtil}
import com.tech.config.ApplicationConfig
import com.tech.process.nodeProcess._
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession

import java.sql.{Connection, DriverManager}
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

class LoadResourceManager extends Serializable {

  var activityBroadcast: Broadcast[ArrayBuffer[String]] = _
  var behaviorMapBroadcast: Broadcast[mutable.HashMap[String, String]] = _
  var startNodeMapBroadcast: Broadcast[mutable.HashMap[String, String]] = _
  var nextNodeMapBroadcast: Broadcast[mutable.HashMap[String, NextNode]] = _
  var groupMapBroadcast: Broadcast[mutable.HashMap[String, mutable.Set[String]]] = _
  var behaviorNodeMapBroadcast: Broadcast[mutable.HashMap[String, BehaviorNode]] = _
  var testNodeMapBroadcast: Broadcast[mutable.HashMap[String, TestNode]] = _
  var waitNodeMapBroadcast: Broadcast[mutable.HashMap[String, WaitNode]] = _

  var eventKeySetBroadcast: Broadcast[mutable.HashSet[String]] = _

  var dimDataGroupDetailMapBroadcast: Broadcast[mutable.HashMap[String, ArrayBuffer[String]]] = _

  var retaileventSchemaMapBroadcast: Broadcast[mutable.HashMap[String, String]] = _
  var customerSchemaMapBroadcast: Broadcast[mutable.HashMap[String, String]] = _
  var columnTypeMapBroadcast: Broadcast[mutable.HashMap[String, String]] = _


  def unPersist(): Unit = {

    activityBroadcast.unpersist(true)
    behaviorMapBroadcast.unpersist(true)
    startNodeMapBroadcast.unpersist(true)
    nextNodeMapBroadcast.unpersist(true)
    groupMapBroadcast.unpersist(true)
    behaviorNodeMapBroadcast.unpersist(true)
    testNodeMapBroadcast.unpersist(true)
    waitNodeMapBroadcast.unpersist(true)
    eventKeySetBroadcast.unpersist(true)

  }

  def unPersistDimDataGroupDetail(): Unit = {

    dimDataGroupDetailMapBroadcast.unpersist(true)

  }

  def load(spark: SparkSession) {

    //每次加载广播变量时重新建立Mysql链接
    val connection: Connection = DriverManager.getConnection(ApplicationConfig.URL_MYSQL)

    val activities = MysqlUtil.getActivity(connection)
    println("activities:" + activities)
    println(activities.isEmpty)

    val behaviorMap = MysqlUtil.getBehaviorMap(connection, activities)

    val startNodes = MysqlUtil.getStartNodes(connection, activities)
    val nextNodes = MysqlUtil.getNextNodes(connection, activities)
    val groups = MysqlUtil.getGroups(connection, activities)
    val behaviorNodes = MysqlUtil.getBehaviorNodes(connection, activities)
    val testNodes = MysqlUtil.getTestNodes(connection, activities)
    val waitNodes = MysqlUtil.getWaitNodes(connection, activities)


    activityBroadcast = spark.sparkContext.broadcast(activities)
    behaviorMapBroadcast = spark.sparkContext.broadcast(behaviorMap)
    startNodeMapBroadcast = spark.sparkContext.broadcast(startNodes)
    nextNodeMapBroadcast = spark.sparkContext.broadcast(nextNodes)
    groupMapBroadcast = spark.sparkContext.broadcast(groups)
    behaviorNodeMapBroadcast = spark.sparkContext.broadcast(behaviorNodes)
    testNodeMapBroadcast = spark.sparkContext.broadcast(testNodes)
    waitNodeMapBroadcast = spark.sparkContext.broadcast(waitNodes)

    println("activityBroadcast.value:" + activityBroadcast.value)

    connection.close()
  }

  def load4D365(spark: SparkSession) {

    val activities = KuduUtil.getActivity4D365
    println("activities:" + activities)
    println(activities.isEmpty)

    val behaviorMap = KuduUtil.getBehaviorMap4D365(activities)

    val startNodes = KuduUtil.getStartNodes4D365(activities)
    val nextNodes = KuduUtil.getNextNodes4D365(activities)
    val groups = KuduUtil.getGroups4D365(activities)

    val behaviorNodes = KuduUtil.getBehaviorNodes4D365(activities)
    val testNodes = KuduUtil.getTestNodes4D365(activities)
    val waitNodes = KuduUtil.getWaitNodes4D365(activities)

    val eventKeys = KuduUtil.getEventKeys4D365(activities)

    activityBroadcast = spark.sparkContext.broadcast(activities)
    behaviorMapBroadcast = spark.sparkContext.broadcast(behaviorMap)
    startNodeMapBroadcast = spark.sparkContext.broadcast(startNodes)
    nextNodeMapBroadcast = spark.sparkContext.broadcast(nextNodes)
    groupMapBroadcast = spark.sparkContext.broadcast(groups)
    behaviorNodeMapBroadcast = spark.sparkContext.broadcast(behaviorNodes)
    testNodeMapBroadcast = spark.sparkContext.broadcast(testNodes)
    waitNodeMapBroadcast = spark.sparkContext.broadcast(waitNodes)
    eventKeySetBroadcast = spark.sparkContext.broadcast(eventKeys)

    println("activityBroadcast.value:" + activityBroadcast.value)
  }

  def loadDimDataGroupDetail(spark: SparkSession): Unit = {

    val dimDataGroupDetail = KuduUtil.getDimDataGroupDetail

    dimDataGroupDetailMapBroadcast = spark.sparkContext.broadcast(dimDataGroupDetail)

  }


  def loadOnlyOnce(spark: SparkSession): Unit = {

    val retaileventSchemaMap = KuduUtil.getTableSchema("retailevent")
    val customerSchemaMap = KuduUtil.getTableSchema("customer")

    retaileventSchemaMapBroadcast = spark.sparkContext.broadcast(retaileventSchemaMap)
    customerSchemaMapBroadcast = spark.sparkContext.broadcast(customerSchemaMap)

    val columnTypeMap = KuduUtil.getColumnTypeMap

    columnTypeMapBroadcast = spark.sparkContext.broadcast(columnTypeMap)

  }
}
